package cn.zf233.xcloud.service.core;

import cn.zf233.xcloud.entity.File;
import cn.zf233.xcloud.util.OSSUtil;
import cn.zf233.xcloud.util.RedisUtil;
import com.aliyun.oss.OSS;
import com.aliyun.oss.model.PartETag;
import org.springframework.beans.factory.DisposableBean;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;

import java.io.IOException;
import java.io.Serializable;
import java.util.ArrayList;
import java.util.LinkedList;
import java.util.List;
import java.util.Queue;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.ReentrantLock;

/**
 * Created by zf233 on 2021/2/23
 * XCloud文件上传功能实现的核心类
 */
public class PersistenceService implements DisposableBean {

    private final int consumerSize;
    private int maxTaskSizeOfConsumer = 20;
    volatile private boolean flag = true;

    private ThreadPoolTaskExecutor threadPool;
    private OSSUtil ossUtil;
    private RedisUtil redisUtil;
    private final MemoryCoordinate memory;
    private final IFailFileRemove iFailFileRemove;

    private List<LinkedList<Product>> shareProductList;
    private final ReentrantLock productLock = new ReentrantLock();
    private List<ReentrantLock> consumerLockList = new ArrayList<>();
    private List<Condition> consumerConditionList = new ArrayList<>();

    public void production(Queue<Product> productQueue) {
        productLock.lock();
        boolean flag = true;
        try {
            while (flag) {
                for (int no = 0; no < consumerSize; no++) {
                    ReentrantLock lock = consumerLockList.get(no);
                    if (lock == null || !lock.tryLock()) {
                        continue;
                    }
                    try {
                        LinkedList<Product> productLink = shareProductList.get(no);
                        while (productLink.size() < maxTaskSizeOfConsumer) {
                            Product file = productQueue.poll();
                            if (file != null) {
                                productLink.add(file);
                            } else {
                                flag = false;
                                break;
                            }
                        }
                        consumerConditionList.get(no).signal();
                    } finally {
                        if (lock.isHeldByCurrentThread()) {
                            lock.unlock();
                        }
                    }
                    if (!flag) {
                        break;
                    }
                }
            }
        } finally {
            if (productLock.isHeldByCurrentThread()) {
                productLock.unlock();
            }
        }
    }

    private void consumption(int no) {
        ReentrantLock lock = consumerLockList.get(no);
        lock.lock();
        try {
            LinkedList<Product> productLink = shareProductList.get(no);
            OSS ossClient = ossUtil.getOSSClient();
            while (!productLink.isEmpty()) {
                Product product = productLink.poll();
                String fileKey = product.getRedisKey();
                File file = product.getFile();
                String randomName = file.getRandomFileName();
                if (redisUtil.cacheKeyExists(fileKey)) { // key存在
                    byte[] cache;
                    if (product.getShardKeyList() == null) {
                        cache = redisUtil.getFileCache(fileKey); // get
                        try {
                            ossUtil.upload(ossClient, randomName, file.getOldFileName(), cache);
                        } catch (IOException e) {
                            iFailFileRemove.remove(file.getRandomFileName());
                        }
                    } else {
                        String uploadId = ossUtil.initShard(ossClient, randomName, file.getOldFileName(), file.getFileSize());
                        boolean complete = true;
                        int i = 0;
                        List<PartETag> partETagList = new ArrayList<>();
                        for (String shardKey : product.getShardKeyList()) {
                            if (!redisUtil.cacheKeyExists(fileKey, shardKey)) {
                                ossUtil.cancelShard(ossClient, randomName, uploadId);
                                iFailFileRemove.remove(file.getRandomFileName());
                                complete = false;
                                break;
                            }
                            cache = redisUtil.getFileCache(fileKey, shardKey);
                            PartETag partETag = ossUtil.uploadShard(ossClient, randomName, uploadId, cache, ++i);
                            partETagList.add(partETag);
                        }
                        if (complete) {
                            ossUtil.submitShard(ossClient, randomName, uploadId, partETagList);
                        }
                    }
                    redisUtil.removeFileCache(fileKey);
                } else {
                    iFailFileRemove.remove(file.getRandomFileName());
                }
                for (; ; ) {
                    if (memory.release(memory.left(), file.getFileSize())) {
                        break;
                    }
                }
            }
            ossClient.shutdown();
            consumerConditionList.get(no).await();
        } catch (InterruptedException e) {
            e.printStackTrace();
        } finally {
            if (lock.isHeldByCurrentThread()) {
                lock.unlock();
            }
        }
    }

    public void init() {
        shareProductList = new ArrayList<>(consumerSize);
        consumerLockList = new ArrayList<>(consumerSize);
        consumerConditionList = new ArrayList<>(consumerSize);
        for (int no = 0; no < consumerSize; no++) {
            ReentrantLock lock = new ReentrantLock();
            consumerLockList.add(lock);
            consumerConditionList.add(lock.newCondition());
            shareProductList.add(new LinkedList<>());
            threadPool.submit(new ConsumerRunnable(no));
        }
    }

    @Override
    public void destroy() {
        flag = false;
        for (int no = 0; no < consumerSize; no++) {
            ReentrantLock lock = consumerLockList.get(no);
            lock.lock();
            try {
                consumerConditionList.get(no).signal();
            } finally {
                if (lock.isHeldByCurrentThread()) {
                    lock.unlock();
                }
            }
        }
    }

    public PersistenceService(Integer consumerSize, MemoryCoordinate memory, IFailFileRemove iFailFileRemove) {
        this.consumerSize = consumerSize;
        this.memory = memory;
        this.iFailFileRemove = iFailFileRemove;
    }

    public void setMaxTaskSizeOfConsumer(Integer maxTaskSizeOfConsumer) {
        this.maxTaskSizeOfConsumer = maxTaskSizeOfConsumer;
    }

    public void setThreadPool(ThreadPoolTaskExecutor threadPool) {
        this.threadPool = threadPool;
    }

    public void setOssUtil(OSSUtil ossUtil) {
        this.ossUtil = ossUtil;
    }

    public void setRedisUtil(RedisUtil redisUtil) {
        this.redisUtil = redisUtil;
    }

    private class ConsumerRunnable implements Runnable {

        private final int no;

        public ConsumerRunnable(Integer no) {
            this.no = no;
        }

        @Override
        public void run() {
            while (flag) {
                consumption(no);
            }
        }
    }

    public static class Product implements Serializable {

        private static final long serialVersionUID = 1632518934309193906L;
        private final File file;
        private final String redisKey;
        private List<String> shardKeyList;

        public Product(File file, String redisKey) {
            this.file = file;
            this.redisKey = redisKey;
        }

        public File getFile() {
            return file;
        }

        public String getRedisKey() {
            return redisKey;
        }

        public List<String> getShardKeyList() {
            return shardKeyList;
        }

        public void setShardKeyList(List<String> shardKeyList) {
            this.shardKeyList = shardKeyList;
        }
    }

    public interface IFailFileRemove {
        void remove(String fileName);
    }
}